package com.xiaoxu.spark_base.example.RDD

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Rdd_Programming_Guide {

  def main(args: Array[String]): Unit = {

    /*
    Spark的API在很大程度上依赖于在驱动程序中传递函数以在集群上运行。
    有两种建议的方法可以做到这一点：

      1.匿名函数语法，可用于短代码。
      2.全局单例对象中的静态方法。例如，您可以定义object MyFunctions
          然后传递MyFunctions.func1
     */
    val conf = new SparkConf().setAppName("appName").setMaster("local")
    val sc = new SparkContext(conf)

    val lines = sc.textFile("data.txt")
    //1.匿名函数语法，可用于短代码。
    val lineLengths = lines.map(x => x.length)
    //2.全局单例对象中的静态方法。
    lines.map(x => MyFunctions.func1(x))


    /**
      * 在这里，如果创建一个新的MyClass实例，并调用doStuff就可以了，
      * map里面有引用的 func1方法是的MyClass实例，所以需要把整个对象发送到群集。
      * 它类似于写作rdd.map(x => this.func1(x))。
      */
    val myClass1 = new MyClass1
    myClass1.func1("hello")


    /**
      * todo 理解闭包
      * 本地变量（单机变量与rdd混用问题）
      */
    //本地变量
    var counter = 0
    var rdd = sc.parallelize("data.txt")

    // Wrong: Don't do this!!
    rdd.foreach(x => counter += x)

    println("Counter value: " + counter)


    /**
      * 打印RDD的元素
      */

    /*
    在一台机器上，这将生成预期的输出并打印所有RDD的元素。
    但是，在cluster模式下，stdout执行程序调用的输出现在写入执行stdout程序，
    而不是驱动程序上的输出，因此stdout驱动程序不会显示这些！
     */
    rdd.foreach(println)
    rdd.map(println)

    //正确做法
    //collect()将整个RDD提取到一台机器上，这会导致驱动程序内存不足。
    rdd.collect().foreach(println)
    rdd.take(100).foreach(println)


    /**
      * PairRDD
      * Extra functions available on RDDs of (key, value) pairs through an implicit conversion.
      *
      * @Experimental todo 引出版本升级经验问题
      *               Experimental API's might change or be removed in minor versions of Spark, or be adopted as
      *               first-class Spark API's.
      */


    val rdd1 = sc.textFile("data.txt").persist()
    val pairs = rdd1.map(s => (s, 1))
    val counts = pairs.reduceByKey((a, b) => a + b)


    /**
      * 共享变量
      * 广播变量
      *
      */
    val data = Array(1, 2, 3, 4, 5)
    val distData = sc.parallelize(data)

    val broadcastVar = sc.broadcast(Array(1, 2, 3))
    distData.foreachPartition(
      records => records.map {
        //每台节点在本地使用广播变量
        record => record + broadcastVar.value(0)
      }
    )


  }

  object MyFunctions {
    def func1(s: String): Int = {
      s.length
    }
  }


  class MyClass1 {

    //there has thousands of variable

    //there has many functions


    def func1(s: String): String = {
      s
    }

    def doStuff(rdd: RDD[String]): RDD[String] = {
      rdd.map(func1)
    }
  }


  /*
  相当于写作rdd.map(x => this.field + x)，它引用了所有this。
  要避免此问题，最简单的方法是复制field到本地变量而不是从外部访问它：
   */

  class MyClass2 {
    val field = "Hello"

    def doStuff(rdd: RDD[String]): RDD[String] = {
      rdd.map(x => field + x)
    }
  }


  class MyClass3 {
    val field = "Hello"

    def doStuff(rdd: RDD[String]): RDD[String] = {
      val field_ = this.field
      rdd.map(x => field_ + x)
    }
  }

}
